package org.qiyu.live.user.provider.Service.Impl;


import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.shaded.com.google.common.collect.Maps;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.idea.qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import org.qiyu.live.common.interfaces.utils.ConvertBeanUtils;
import org.qiyu.live.user.constants.CacheAsyncDeleteCode;
import org.qiyu.live.user.constants.UserPeoviderTopicNames;
import org.qiyu.live.user.dto.UserCacheAsyncDeleteDTO;
import org.qiyu.live.user.dto.UserDTO;
import org.qiyu.live.user.provider.Service.IUserService;
import org.qiyu.live.user.provider.dao.mapper.UserMapper;
import org.qiyu.live.user.provider.dao.po.User;
import org.qiyu.live.user.provider.dao.po.UserPO;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @ClassName: UserServiceImpl
 * @Description: TODO
 * @Author zhangguangyan
 * @Date 2023/07/23/ 15:16:02
 * @Version 1.0
 */
@Service
public class UserServiceImpl implements IUserService {
    @Resource
    private UserMapper userMapper;
    @Resource
    private RedisTemplate<String, UserDTO> redisTemplate;

    @Resource
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;
    @Resource
    private MQProducer mqProducer;



    @Override
    public UserDTO getByUserId(Long userId) {
        if (userId == null) {
            return null;
        }
        String key =userProviderCacheKeyBuilder.buildUserInfoKey(userId);
        UserDTO userDTO = redisTemplate.opsForValue().get(key);
        if (userDTO != null) {
            return userDTO;
        }
        userDTO  =ConvertBeanUtils.convert(userMapper.selectById(userId), UserDTO.class);
        if (userDTO !=null){
            redisTemplate.opsForValue().set(key,userDTO,30, TimeUnit.MINUTES);
        }
        return userDTO;
    }


    @Override
    public boolean updateUserInfo(UserDTO userDTO) {
        if (userDTO == null || userDTO.getUserId() == null) {
            return false;
        }
        int updateStatus  = userMapper.updateById(ConvertBeanUtils.convert(userDTO, UserPO.class));
        if (updateStatus >0){
            String key =userProviderCacheKeyBuilder.buildUserInfoKey(userDTO.getUserId());
            redisTemplate.delete(key);
            HashMap map = new HashMap() {{
                put("userId", userDTO.getUserId());
            }};
            UserCacheAsyncDeleteDTO.builder().code(CacheAsyncDeleteCode.USER_INFO_DELETE.getCode()).json(JSON.toJSONString(map)).build();
        }
        Message message = new Message();
        message.setBody(JSON.toJSONString(userDTO).getBytes());
        message.setTopic(UserPeoviderTopicNames.CACHE_ASYNC_DELETE_TOPIC);
        //延迟级别，1代表延迟一秒发送
        message.setDelayTimeLevel(1);
        try {
//延迟疫苗进行缓存的二次删除
            mqProducer.send(message);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return true;
    }

    @Override
    public boolean insertOne(UserDTO userDTO) {
        if (userDTO == null || userDTO.getUserId() == null) {
            return false;
        }
        userMapper.insert(ConvertBeanUtils.convert(userDTO, UserPO.class));
        return true;
    }

    @Override
    public Map<Long, UserDTO> batchQueryUserInfo(List<Long> userIdList) {
        if (CollectionUtils.isEmpty(userIdList)) {
            return Maps.newHashMap();
        }
        //多线程查询 替换union all
        userIdList = userIdList.stream().filter(e -> e > 1000).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(userIdList)) {
            return Maps.newHashMap();
        }

        //redis
        List<String> keyList = new ArrayList<>();
        userIdList.forEach(e -> {
            // 重新定义key值
            keyList.add(userProviderCacheKeyBuilder.buildUserInfoKey(e));
        });
//从redis中查看是否有数据，如果有数据则直接返回 如果没有数据则从数据库中查询
        List<UserDTO> userDTOList = redisTemplate.opsForValue().multiGet(keyList).stream().filter(x -> x != null).collect(Collectors.toList());
        if (userDTOList.size() == userIdList.size() && CollectionUtils.isEmpty(userDTOList)) {
            return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId, e -> e));
        }
        List<Long> userIdInCacheList = userDTOList.stream().map(UserDTO::getUserId).collect(Collectors.toList());
        List<Long> userIdNotInCacheList = userIdList.stream().filter(e -> !userIdInCacheList.contains(e)).collect(Collectors.toList());
        Map<Long, List<Long>> userIdMap = userIdNotInCacheList.stream().collect(Collectors.groupingBy(userId -> userId % 100));        List<UserDTO> dbQueryResult = new CopyOnWriteArrayList<>();

        userIdMap.values().parallelStream().forEach(queryUserIdList -> {
            dbQueryResult.addAll(ConvertBeanUtils.convertList(userMapper.selectBatchIds(queryUserIdList), UserDTO.class));
        });
        if (!CollectionUtils.isEmpty(dbQueryResult)) {
            Map<String, UserDTO> collect = dbQueryResult.stream().collect(Collectors.toMap(userDto -> userProviderCacheKeyBuilder.buildUserInfoKey(userDto.getUserId()), e -> e));
            redisTemplate.opsForValue().multiSet(collect);
            //管道批量传输命令  减少网络io开销
            redisTemplate.executePipelined(new SessionCallback<Object>() {
                @Override
                public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                    for (String rediskey :collect.keySet()) {
                        operations.expire((K) rediskey,createRandomExpireTime(),TimeUnit.SECONDS);

                    }
                    return null;
                }
            });
            userDTOList.addAll(dbQueryResult);
        }

        return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId, e -> e));
    }

    private  int createRandomExpireTime(){
        int i = ThreadLocalRandom.current().nextInt(1000);
        return i+60*30;
    }

    public static void main(String[] args) {
        List<String> list = new ArrayList<>();
        list.add("张三");
        list.add("李四");
        list.add("王五");
        list.add("马六");

        List<User> listUser = new ArrayList<>();
        ;
        listUser.add(new User().builder().id(1).name("里斯1").age(23).build());
        listUser.add(new User().builder().id(2).name("里斯2").age(24).build());
        listUser.add(new User().builder().id(3).name("里斯3").age(25).build());
        listUser.add(new User().builder().id(4).name("里斯4").age(26).build());
//        List<Object> collect = listUser.stream().map( User::getName,User::getId).collect(Collectors.toList());
//        Map<Integer, String> listUserMap = listUser.stream().collect(Collectors.toMap(User::getId,User::getName,User::getAgea));

    }
}
